package cn.wolfcode._01hello;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 消费者
 */
public class Consumer {
    /*public static void main(String[] args) throws Exception {
        // 实例化消息生产者Producer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("wolfcode_consumer");
        // 设置NameServer的地址
        consumer.setNamesrvAddr("192.168.204.128:9876"); //服务所在的电脑ip
        // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息  --监听该主题消息
        consumer.subscribe("TopicTest", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for(MessageExt messageExt : msgs){
                    System.out.println("消费线程:"+Thread.currentThread().getName()+",消息ID:"+messageExt.getMsgId()+",消息内容:"+new String(messageExt.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }*/

   //集群和广播模式
   // 延时消息 -------------------------------------------------
   public static void main(String[] args) throws Exception {
       // 定义消息的消费者
       DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("wolfcode_consumer-jiqun");
       // 设置 nameServer地址
       consumer.setNamesrvAddr("192.168.204.128:9876");
       // 定义消费者模式 -- 集群模式（默认）和 广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);// 广播模式

       // 监听 改主题消息（注意要和生产者一致）
       consumer.subscribe("jiqunTopic", "*"); //* 表示监听jqunTopic主题的所有消息
       // 设置消息监听器，服务器把消息推送给我们，消费消息
       consumer.setMessageListener(new MessageListenerConcurrently() {
           @Override
           public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
               // 消息处理业务逻辑
               for (MessageExt msg : list) {
                   // 可以看到每个queue有唯一的consume线程来消费, 订单对每个queue(分区)有序
                   System.out.println("处理线程=" + Thread.currentThread().getName() + "，消息内容=" + new String(msg.getBody()));
               }
               // 告诉消息中间件，消息处理情况
               return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
           }
       });
       // 启动
       consumer.start();
   }
}